
const mqtt = require('mqtt');
const http = require('http');
const WebSocket = require('ws');
const axios = require('axios');
const fs = require('fs');
const path = require('path');

// 读取配置文件
let configData = {};
try {
  const configPath = path.join(__dirname, 'config.json');
  const configContent = fs.readFileSync(configPath, 'utf8');
  configData = JSON.parse(configContent);
  console.log('成功读取配置文件:', configPath);
} catch (error) {
  console.error('读取配置文件失败，使用默认配置:', error.message);
  // 使用默认配置作为后备
  configData = {
    DINGTALK_ROBOT: {
      webhook: 'https://oapi.dingtalk.com/robot/send?access_token=996633eb7f07b8e060c0611eeeb8b1a2ad7d51f33e71397742483d9a103c2908',
      keywords: ['VBAT']
    },
    VOLTAGE_THRESHOLDS: {
      lower: 4.00,
      upper: 4.30
    }
  };
}

// 全局配置常量
const CONFIG = {
  // 钉钉机器人配置
  DINGTALK_ROBOT: {
    webhook: configData.DINGTALK_ROBOT?.webhook || 'https://oapi.dingtalk.com/robot/send?access_token=996633eb7f07b8e060c0611eeeb8b1a2ad7d51f33e71397742483d9a103c2908',
    // 钉钉机器人安全设置关键词（从配置文件读取）
    keywords: configData.DINGTALK_ROBOT?.keywords || ['VBAT']
  },
  // 电压警告阈值配置（单位：V，从配置文件读取）
  VOLTAGE_THRESHOLDS: {
    lower: configData.VOLTAGE_THRESHOLDS?.lower || 4.00, // 下限阈值
    upper: configData.VOLTAGE_THRESHOLDS?.upper || 4.30  // 上限阈值
  },
  // 系统配置（从配置文件读取）
  SYSTEM_CONFIG: {
    ALERT_COOLDOWN_TIME: configData.SYSTEM_CONFIG?.ALERT_COOLDOWN_TIME || 0, // 告警冷却时间（毫秒），默认0（无冷却）
    maxHistorySize: configData.SYSTEM_CONFIG?.maxHistorySize || 15 // 最大历史记录数，默认15条
  }
};

// 为了保持向后兼容性，保留DINGTALK_ROBOT_CONFIG引用
const DINGTALK_ROBOT_CONFIG = {
  webhook: CONFIG.DINGTALK_ROBOT.webhook,
  voltageThresholds: CONFIG.VOLTAGE_THRESHOLDS,
  keywords: CONFIG.DINGTALK_ROBOT.keywords
};

// 数据持久化配置
const DATA_FILE_PATH = path.join(__dirname, 'voltage_history.json'); // 数据持久化文件路径
const DATA_SAVE_INTERVAL = 5000; // 数据保存间隔（毫秒）

// 存储最近N条电压数据
let recentVoltageData = [];

// 保存电压历史数据到文件
function saveVoltageDataToFile() {
  try {
    const dataToSave = {
      timestamp: Date.now(),
      records: recentVoltageData,
      totalRecords: recentVoltageData.length
    };
    fs.writeFileSync(DATA_FILE_PATH, JSON.stringify(dataToSave, null, 2));
    logger.debug('电压历史数据已保存到文件', { recordCount: recentVoltageData.length });
  } catch (error) {
    logger.error('保存电压历史数据到文件失败:', { error: error.message });
  }
}

// 从文件加载电压历史数据
function loadVoltageDataFromFile() {
  try {
    if (fs.existsSync(DATA_FILE_PATH)) {
      const fileContent = fs.readFileSync(DATA_FILE_PATH, 'utf8');
      const savedData = JSON.parse(fileContent);
      if (savedData.records && Array.isArray(savedData.records)) {
        recentVoltageData = savedData.records;
        logger.info(`从文件加载了${recentVoltageData.length}条电压历史数据`);
      }
    }
  } catch (error) {
    logger.error('从文件加载电压历史数据失败:', { error: error.message });
    // 发生错误时，使用空数组作为后备
    recentVoltageData = [];
  }
}

// 日志记录函数，用于格式化输出详细日志
function log(level, message, context = {}) {
  // 使用北京时间输出日志
  const beijingTime = new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' });
  console[level](`[${beijingTime}] [${level.toUpperCase()}] ${message}`, context);
}

// 简化日志函数
const logger = {
  debug: (message, context = {}) => log('debug', message, context),
  info: (message, context = {}) => log('info', message, context),
  warn: (message, context = {}) => log('warn', message, context),
  error: (message, context = {}) => log('error', message, context)
};

// 初始化时加载历史数据
loadVoltageDataFromFile();

// 按需保存数据 - 移除了定时保存，改为在添加新数据时保存

// 用于记录已经发送的告警，避免重复推送
const alertHistory = new Map();
const ALERT_COOLDOWN_TIME = CONFIG.SYSTEM_CONFIG.ALERT_COOLDOWN_TIME; // 告警冷却时间（毫秒），从配置文件读取

// 最大历史记录数，从配置文件读取
const maxHistorySize = CONFIG.SYSTEM_CONFIG.maxHistorySize;

// 存储设备登录记录
const deviceLoginRecords = new Map();
const maxLoginRecords = 50;

// 存储所有需要持久化的重要消息
const importantMessages = [];
const maxImportantMessages = 100;

// 发送钉钉机器人消息
async function sendDingTalkMessage(title, text) {
  logger.debug('准备发送钉钉机器人消息', { title: title.substring(0, 50) });
  try {
    // 确保消息包含必要的关键词
    let processedText = text;
    const hasKeyword = CONFIG.DINGTALK_ROBOT.keywords.some(keyword => 
      title.includes(keyword) || text.includes(keyword)
    );
    
    if (!hasKeyword && CONFIG.DINGTALK_ROBOT.keywords.length > 0) {
      // 如果不包含关键词，自动添加到消息开头
      const keywordsText = CONFIG.DINGTALK_ROBOT.keywords.join('、');
      processedText = `${keywordsText}：\n\n${text}`;
      logger.debug('自动添加关键词到消息', { addedKeywords: keywordsText });
    }
    
    const message = {
      msgtype: 'markdown',
      markdown: {
        title: title,
        text: processedText
      }
    };

    const response = await axios.post(CONFIG.DINGTALK_ROBOT.webhook, message, {
      headers: {
        'Content-Type': 'application/json'
      }
    });

    logger.info('钉钉机器人消息发送成功', { 
      title: title.substring(0, 50),
      responseStatus: response.status,
      responseData: response.data
    });
    return response.data;
  } catch (error) {
    logger.error('钉钉机器人消息发送失败', {
      error: error.message,
      title: title.substring(0, 50),
      stack: error.stack
    });
    return null;
  }
}

// 从数据中提取电压值的辅助函数
function extractVoltageData(data) {
  logger.debug('开始提取电压数据', { dataSample: JSON.stringify(data).substring(0, 200) });
  let vbatValue = null;
  let sourceField = 'unknown';
  
  // 检查数据是否为数组格式
  if (Array.isArray(data)) {
    logger.debug('检测到数组格式数据，开始遍历处理', { arrayLength: data.length });
    
    // 存储所有有效的电压值
    const validVoltages = [];
    
    // 遍历数组中的每个对象
    for (let i = 0; i < data.length; i++) {
      const item = data[i];
      let itemVoltage = null;
      
      // 尝试从当前对象中提取电压数据
      if (item.vbat !== undefined) {
        itemVoltage = parseFloat(item.vbat);
        sourceField = `array[${i}].vbat`;
        logger.debug(`从数组元素vbat字段提取电压数据`, { index: i, value: itemVoltage });
      } else if (item.voltage !== undefined) {
        itemVoltage = parseFloat(item.voltage);
        sourceField = `array[${i}].voltage`;
        logger.debug(`从数组元素voltage字段提取电压数据`, { index: i, value: itemVoltage });
      }
      
      // 如果提取到有效电压值，添加到有效电压数组
      if (itemVoltage !== null && !isNaN(itemVoltage)) {
        // 自动检测单位并转换（如果大于1000，假设是毫伏）
        if (itemVoltage > 1000) {
          itemVoltage = itemVoltage / 1000;
          logger.debug(`数组元素电压数据单位转换: ${itemVoltage * 1000}mV -> ${itemVoltage}V`);
        }
        validVoltages.push(itemVoltage);
      }
    }
    
    // 如果有有效的电压值，取最新的一个（数组最后一个）
    if (validVoltages.length > 0) {
      vbatValue = validVoltages[validVoltages.length - 1];
      sourceField = 'array';
      logger.debug(`成功从数组数据中提取电压值`, { 
        value: vbatValue, 
        validCount: validVoltages.length, 
        totalItems: data.length 
      });
    }
    
    // 对于数组数据，直接返回结果，不再执行后续的单个对象处理逻辑
    if (validVoltages.length > 0) {
      logger.debug('数组电压数据提取完成', { result: { vbat: vbatValue, sourceField } });
      return { vbat: vbatValue, sourceField };
    }
  }
  
  // 尝试从常见字段中提取电压数据
  if (data.vbat !== undefined) {
    vbatValue = parseFloat(data.vbat);
    sourceField = 'vbat';
    logger.debug('从vbat字段提取电压数据', { value: vbatValue });
  } else if (data.voltage !== undefined) {
    vbatValue = parseFloat(data.voltage);
    sourceField = 'voltage';
    logger.debug('从voltage字段提取电压数据', { value: vbatValue });
  } else if (data.value !== undefined) {
    vbatValue = parseFloat(data.value);
    sourceField = 'value';
    logger.debug('从value字段提取电压数据', { value: vbatValue });
  } else if (data.payload && typeof data.payload === 'object') {
    // 检查嵌套payload对象
    if (data.payload.vbat !== undefined) {
      vbatValue = parseFloat(data.payload.vbat);
      sourceField = 'payload.vbat';
      logger.debug('从payload.vbat字段提取电压数据', { value: vbatValue });
    } else if (data.payload.voltage !== undefined) {
      vbatValue = parseFloat(data.payload.voltage);
      sourceField = 'payload.voltage';
      logger.debug('从payload.voltage字段提取电压数据', { value: vbatValue });
    }
  } else if (data.payload && typeof data.payload === 'string') {
    // 尝试解析字符串格式的payload
    try {
      const parsedPayload = JSON.parse(data.payload);
      if (parsedPayload.vbat !== undefined) {
        vbatValue = parseFloat(parsedPayload.vbat);
        sourceField = 'parsedPayload.vbat';
        logger.debug('从parsedPayload.vbat字段提取电压数据', { value: vbatValue });
      } else if (parsedPayload.voltage !== undefined) {
        vbatValue = parseFloat(parsedPayload.voltage);
        sourceField = 'parsedPayload.voltage';
        logger.debug('从parsedPayload.voltage字段提取电压数据', { value: vbatValue });
      }
    } catch (e) {
      // 解析失败，忽略
      logger.warn('无法解析payload字符串', { payloadSample: data.payload.substring(0, 100), error: e.message });
    }
  }
  
  // 自动检测单位并转换（如果大于1000，假设是毫伏）
  if (vbatValue !== null && !isNaN(vbatValue) && vbatValue > 1000) {
    logger.debug(`检测到毫伏单位数据: ${vbatValue}mV，转换为伏特`);
    vbatValue = vbatValue / 1000;
  }
  
  logger.debug('电压数据提取完成', { result: { vbat: vbatValue, sourceField } });
  return { vbat: vbatValue, sourceField };
}

// 检查是否需要发送电压告警
async function checkVoltageAlert(vbat) {
  logger.debug('开始检查电压告警', { voltage: vbat });
  const voltage = parseFloat(vbat);
  if (isNaN(voltage)) {
    logger.warn('无效的电压值，跳过告警检查', { value: vbat });
    return;
  }

  // 改进告警键的设计，使用更精确的标识
  const alertType = voltage < CONFIG.VOLTAGE_THRESHOLDS.lower ? 'low_voltage' : 
                   voltage > CONFIG.VOLTAGE_THRESHOLDS.upper ? 'high_voltage' : 'normal';
  
  // 只有当电压超出阈值时才检查冷却时间
  if (alertType !== 'normal') {
    const alertKey = `${alertType}_alert`;
    const lastAlertTime = alertHistory.get(alertKey) || 0;
    const now = Date.now();

    // 检查是否在冷却时间内
    if (now - lastAlertTime < CONFIG.SYSTEM_CONFIG.ALERT_COOLDOWN_TIME) {
      logger.debug('电压告警在冷却时间内，跳过告警', {
        alertKey: alertKey,
        lastAlertTime: new Date(lastAlertTime).toISOString(),
        remainingCooldown: (CONFIG.SYSTEM_CONFIG.ALERT_COOLDOWN_TIME - (now - lastAlertTime)) / 1000 + 's'
      });
      return;
    }

    let alertTitle = '';
    let alertText = '';
    
    if (alertType === 'low_voltage') {
      alertTitle = '⚠️ 电压过低告警';
      alertText = `### ⚠️ 电压过低告警\n\n` +
                  `**当前电压**: ${voltage.toFixed(3)}V\n` +
                  `**警告阈值**: ${CONFIG.VOLTAGE_THRESHOLDS.lower}V\n` +
                  `**时间**: ${new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })}\n` +
                  `**状态**: 电压已低于安全阈值，请及时处理！`;
    } else if (alertType === 'high_voltage') {
      alertTitle = '⚠️ 电压过高告警';
      alertText = `### ⚠️ 电压过高告警\n\n` +
                  `**当前电压**: ${voltage.toFixed(3)}V\n` +
                  `**警告阈值**: ${CONFIG.VOLTAGE_THRESHOLDS.upper}V\n` +
                  `**时间**: ${new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })}\n` +
                  `**状态**: 电压已超过安全阈值，请及时处理！`;
    }

    // 记录告警发送时间
    alertHistory.set(alertKey, now);
    logger.info('准备发送电压告警', {
      type: alertType,
      voltage: voltage,
      thresholdLow: CONFIG.VOLTAGE_THRESHOLDS.lower,
      thresholdHigh: CONFIG.VOLTAGE_THRESHOLDS.upper
    });
    
    try {
      // 发送告警消息
      const result = await sendDingTalkMessage(alertTitle, alertText);
      logger.info('电压告警消息发送结果', { success: result !== null });
    } catch (error) {
      logger.error('发送电压告警消息时发生异常', {
        error: error.message,
        stack: error.stack
      });
    }
  } else {
    logger.debug('电压在正常范围内，不需要发送告警', {
      voltage: voltage,
      thresholdLow: CONFIG.VOLTAGE_THRESHOLDS.lower,
      thresholdHigh: CONFIG.VOLTAGE_THRESHOLDS.upper
    });
  }

  // 清理过期的告警记录（超过24小时）
  const twentyFourHoursAgo = Date.now() - (24 * 60 * 60 * 1000);
  for (let [key, timestamp] of alertHistory.entries()) {
    if (timestamp < twentyFourHoursAgo) {
      alertHistory.delete(key);
      logger.debug('清理过期的告警记录', { key: key });
    }
  }
}

// 创建HTTP服务器用于健康检查和配置获取
const server = http.createServer((req, res) => {
  logger.debug('收到HTTP请求', { url: req.url, method: req.method });
  
  // 健康检查接口
  if (req.url === '/' && req.method === 'GET') {
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({
      status: 'ok',
      message: 'MQTT proxy server is running',
      timestamp: new Date().toISOString(),
      activeConnections: activeConnections.size,
      recentVoltageDataCount: recentVoltageData.length
    }));
    logger.debug('HTTP健康检查请求处理完成', { url: req.url });
  }
  // 配置获取接口 - 用于小程序获取当前配置
  else if (req.url === '/api/config' && req.method === 'GET') {
    // 返回安全的配置信息，注意不要返回敏感信息
    const safeConfig = {
      DINGTALK_ROBOT: {
        // 不返回完整的webhook，只返回是否配置
        hasWebhook: !!CONFIG.DINGTALK_ROBOT.webhook && CONFIG.DINGTALK_ROBOT.webhook !== 'https://oapi.dingtalk.com/robot/send?access_token=996633eb7f07b8e060c0611eeeb8b1a2ad7d51f33e71397742483d9a103c2908',
        keywords: CONFIG.DINGTALK_ROBOT.keywords
      },
      VOLTAGE_THRESHOLDS: {
        lower: CONFIG.VOLTAGE_THRESHOLDS.lower,
        upper: CONFIG.VOLTAGE_THRESHOLDS.upper
      },
      SYSTEM_CONFIG: {
        ALERT_COOLDOWN_TIME: CONFIG.SYSTEM_CONFIG.ALERT_COOLDOWN_TIME,
        maxHistorySize: CONFIG.SYSTEM_CONFIG.maxHistorySize
      }
    };
    
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({
      status: 'ok',
      config: safeConfig,
      timestamp: new Date().toISOString()
    }));
    logger.debug('配置获取请求处理完成', { url: req.url });
  }
  // 未知路由
  else {
    res.writeHead(404, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({
      status: 'error',
      message: 'Not found'
    }));
    logger.debug('未知HTTP请求处理完成', { url: req.url });
  }
});

// 创建WebSocket服务器
const wss = new WebSocket.Server({ server });

// 存储所有活跃连接
const activeConnections = new Map();

// WebSocket连接处理
wss.on('connection', (ws, req) => {
  // 生成唯一的客户端ID
  const clientId = 'cloud-proxy-' + Date.now() + '-' + Math.random().toString(36).substr(2, 9);
  
  // 获取北京时间的当前时间
  const currentTime = new Date();
  const beijingTimeStr = currentTime.toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' });
  
  logger.info('新的WebSocket客户端连接', {
    clientId: clientId,
    headers: req.headers,
    url: req.url,
    activeConnections: activeConnections.size + 1
  });
  
  // 记录设备登录信息
  const loginRecord = {
    clientId: clientId,
    loginTime: currentTime.getTime(),
    loginTimeBeijing: beijingTimeStr,
    headers: req.headers,
    deviceInfo: 'unknown',
    ip: req.socket.remoteAddress,
    userAgent: req.headers['user-agent']
  };
  
  // 记录连接信息到登录记录中
  deviceLoginRecords.set(clientId, loginRecord);
  
  // 限制登录记录数量
  if (deviceLoginRecords.size > maxLoginRecords) {
    const firstKey = deviceLoginRecords.keys().next().value;
    deviceLoginRecords.delete(firstKey);
  }
  
  // 记录连接信息
  activeConnections.set(clientId, { ws, startTime: Date.now(), loginRecord });
  
  // 发送连接确认消息
  ws.send(JSON.stringify({
    type: 'connection_established',
    clientId: clientId,
    message: 'WebSocket connection established, connecting to MQTT server...',
    timestamp: Date.now()
  }));
  
  // 设置连接超时检测
  const connectionTimeout = setTimeout(() => {
    if (!mqttClient || !mqttClient.connected) {
      logger.warn(`MQTT连接超时 for client ${clientId}`);
      ws.send(JSON.stringify({
        type: 'connection_timeout',
        error: 'Failed to connect to MQTT server within timeout period',
        clientId: clientId,
        timeout: 8000
      }));
    }
  }, 8000); // 8秒超时检测

  // 连接到目标MQTT服务器
  const mqttBroker = 'mqtt://broker.emqx.io';
  logger.info(`开始连接MQTT服务器...`, {
    serverAddress: mqttBroker,
    clientId: clientId
  });
  logger.debug(`MQTT连接参数:`, {
    clientId: `cloud-client-${clientId}`,
    topic: '/robot/zack/vbat',
    clean: true,
    connectTimeout: 5000,
    keepalive: 60
  });
  
  const mqttClient = mqtt.connect(mqttBroker, {
    clientId: `cloud-client-${clientId}`,
    clean: true,
    connectTimeout: 5000,
    keepalive: 60
  });

  // 当MQTT连接成功时
  mqttClient.on('connect', () => {
    logger.info(`MQTT连接成功`, {
      clientId: clientId,
      broker: mqttBroker
    });
    
    // 清除连接超时定时器
    clearTimeout(connectionTimeout);
    
    ws.send(JSON.stringify({
      type: 'connect',
      success: true,
      message: 'MQTT server connected',
      clientId: clientId,
      broker: mqttBroker,
      connectedTime: new Date().toISOString(),
      connectionDetails: {
        cleanSession: true,
        keepalive: 60
      }
    }));
    
    // 订阅默认主题
    mqttClient.subscribe('/robot/zack/vbat', (err) => {
      if (err) {
        logger.error('订阅主题失败', {
          topic: '/robot/zack/vbat',
          clientId: clientId,
          error: err.message
        });
      } else {
        logger.info('成功订阅默认主题', {
          topic: '/robot/zack/vbat',
          clientId: clientId
        });
      }
    });
  });

  // 当收到小程序发送的消息时
  ws.on('message', (message) => {
    logger.debug('收到WebSocket消息', { clientId: clientId, messageLength: message.length });
    try {
      // 检查消息是否为空
      const messageStr = message.toString().trim();
      if (!messageStr) {
        logger.warn(`接收到空消息 from client ${clientId}`);
        ws.send(JSON.stringify({
          type: 'error',
          message: 'Empty message received. Please send valid JSON data.',
          timestamp: Date.now()
        }));
        return;
      }
      
      const data = JSON.parse(messageStr);
      logger.info(`收到来自小程序的消息`, {
        clientId: clientId,
        messageType: data.type,
        messageSize: messageStr.length
      });
      logger.debug('消息内容详情:', { data });
      
      // 检查是否是上线通知
      if (data.type === 'online_notification') {
        logger.info('收到小程序上线通知，准备向MQTT主题发送上线消息', {
          clientId: clientId,
          deviceInfo: data.deviceInfo || 'default'
        });
        
        // 更新登录记录中的设备信息
        if (deviceLoginRecords.has(clientId)) {
          const loginRecord = deviceLoginRecords.get(clientId);
          loginRecord.deviceInfo = data.deviceInfo || {};
          loginRecord.appVersion = data.appVersion || 'unknown';
          loginRecord.onlineTime = Date.now();
          loginRecord.onlineTimeBeijing = new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' });
          deviceLoginRecords.set(clientId, loginRecord);
        }
        
        // 构建上线消息
        const onlineMessage = {
          type: 'device_online',
          device: 'mini_program',
          timestamp: Date.now(),
          timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
          clientId: clientId,
          deviceInfo: data.deviceInfo || {},
          appVersion: data.appVersion || 'unknown'
        };
        
        // 发布上线消息到MQTT主题
        if (mqttClient.connected) {
          mqttClient.publish('/robot/zack/vbat', JSON.stringify(onlineMessage), (err) => {
            if (err) {
              logger.error('发布上线消息失败:', { error: err.message });
            } else {
              logger.info('成功发布上线消息到MQTT主题: /robot/zack/vbat', { onlineMessage });
            }
          });
        } else {
          logger.error('MQTT未连接，无法发布上线消息', {
            clientId: clientId,
            mqttConnected: mqttClient.connected
          });
        }
        
        // 发送当前电压阈值给小程序
          logger.info('向小程序发送当前电压阈值', { clientId: clientId });
          ws.send(JSON.stringify({
            type: 'thresholds_updated',
            data: CONFIG.VOLTAGE_THRESHOLDS,
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
          }));
        
        return;
      }
      
      switch (data.type) {
        case 'publish':
          // 转发发布消息到MQTT服务器
          // 支持标准格式 (topic + payload) 和简化格式 (vbat字段直接作为电压数据)
          logger.debug('处理publish类型消息', { clientId: clientId });
          let topic = data.topic;
          let payload = data.payload;
          
          // 如果消息包含vbat字段但缺少topic和payload，则自动转换为标准格式
          if (!topic && !payload && data.vbat !== undefined) {
            topic = '/robot/zack/vbat';
            payload = JSON.stringify({ vbat: data.vbat });
            logger.debug(`自动转换简化格式消息为标准格式:`, { topic: topic, payloadLength: payload.length });
          }
          
          if (topic && payload) {
            logger.info('转发publish消息到MQTT服务器', {
              clientId: clientId,
              topic: topic,
              payloadLength: payload.length
            });
            mqttClient.publish(topic, payload, data.options || {}, (err) => {
              if (err) {
                logger.error(`Publish error for client ${clientId}:`, { error: err.message, topic: topic });
                ws.send(JSON.stringify({
                  type: 'publish_response',
                  success: false,
                  error: err.message,
                  topic: topic,
                  timestamp: Date.now()
                }));
              } else {
                logger.info(`Publish successful for client ${clientId}`, { topic: topic });
                ws.send(JSON.stringify({
                  type: 'publish_response',
                  success: true,
                  topic: topic,
                  timestamp: Date.now(),
                  payloadLength: payload.length
                }));
              }
            });
          } else {
            logger.warn('缺少必要的publish参数(topic或payload)', { clientId: clientId });
            ws.send(JSON.stringify({
              type: 'error',
              message: 'Missing required parameters: topic and/or payload',
              timestamp: Date.now()
            }));
          }
          break;
          
        case 'subscribe':
          logger.debug('处理subscribe类型消息', { clientId: clientId, topic: data.topic });
          // 转发订阅请求到MQTT服务器
          if (data.topic) {
            logger.info('转发subscribe请求到MQTT服务器', { clientId: clientId, topic: data.topic });
            mqttClient.subscribe(data.topic, data.options || {}, (err) => {
              if (err) {
                logger.error(`Subscribe error for client ${clientId}:`, { error: err.message, topic: data.topic });
                ws.send(JSON.stringify({
                  type: 'subscribe_response',
                  success: false,
                  error: err.message,
                  topic: data.topic,
                  timestamp: Date.now()
                }));
              } else {
                logger.info(`Subscribe successful for client ${clientId}`, { topic: data.topic });
                ws.send(JSON.stringify({
                  type: 'subscribe_response',
                  success: true,
                  topic: data.topic,
                  timestamp: Date.now()
                }));
              }
            });
          } else {
            logger.warn('缺少必要的subscribe参数(topic)', { clientId: clientId });
            ws.send(JSON.stringify({
              type: 'error',
              message: 'Missing required parameter: topic',
              timestamp: Date.now()
            }));
          }
          break;
          
        case 'unsubscribe':
          logger.debug('处理unsubscribe类型消息', { clientId: clientId, topic: data.topic });
          // 转发取消订阅请求到MQTT服务器
          if (data.topic) {
            logger.info('转发unsubscribe请求到MQTT服务器', { clientId: clientId, topic: data.topic });
            mqttClient.unsubscribe(data.topic, data.options || {}, (err) => {
              if (err) {
                logger.error(`Unsubscribe error for client ${clientId}:`, { error: err.message, topic: data.topic });
                ws.send(JSON.stringify({
                  type: 'unsubscribe_response',
                  success: false,
                  error: err.message,
                  topic: data.topic,
                  timestamp: Date.now()
                }));
              } else {
                logger.info(`Unsubscribe successful for client ${clientId}`, { topic: data.topic });
                ws.send(JSON.stringify({
                  type: 'unsubscribe_response',
                  success: true,
                  topic: data.topic,
                  timestamp: Date.now()
                }));
              }
            });
          } else {
            logger.warn('缺少必要的unsubscribe参数(topic)', { clientId: clientId });
            ws.send(JSON.stringify({
              type: 'error',
              message: 'Missing required parameter: topic',
              timestamp: Date.now()
            }));
          }
          break;
          
        case 'ping':
          logger.debug('处理ping类型消息', { clientId: clientId });
          // 处理心跳包，同时转发到MQTT服务器
          ws.send(JSON.stringify({
            type: 'pong',
            timestamp: Date.now(),
            serverTime: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
          }));
          
          // 转发心跳消息到MQTT服务器
          const heartbeatMsg = JSON.stringify({
            type: 'heartbeat',
            ts: Date.now(),
            clientId: clientId,
            mqttConnected: mqttClient.connected
          });
          mqttClient.publish('/robot/zack/vbat', heartbeatMsg, {}, (err) => {
            if (err) {
              logger.error(`Publish heartbeat error for client ${clientId}:`, { error: err.message });
            } else {
              logger.debug(`成功发布心跳消息到MQTT主题: /robot/zack/vbat`, { clientId: clientId });
            }
          });
          break;
          
        case 'voltage_alarm':
          logger.debug('处理voltage_alarm类型消息', { clientId: clientId });
          // 接收小程序发送的电压告警信息，同时支持vbat和bat两种参数名
          const voltageValue = data.vbat !== undefined ? data.vbat : data.bat;
          if (voltageValue && data.alertType) {
            logger.info('接收到电压告警', {
              clientId: clientId,
              voltage: voltageValue,
              alertType: data.alertType,
              thresholdType: data.thresholdType,
              thresholdValue: data.thresholdValue,
              voltageParamName: data.vbat !== undefined ? 'vbat' : 'bat'
            });
            
            // 构建告警消息并发送钉钉通知
            let alertTitle = '';
            let alertText = '';
            
            if (data.alertType === 'low') {
              alertTitle = '⚠️ 电压过低告警';
              alertText = `### ⚠️ 电压过低告警\n\n` +
                          `**当前电压**: ${parseFloat(voltageValue).toFixed(3)}V\n` +
                          `**警告阈值**: ${parseFloat(data.thresholdValue).toFixed(3)}V\n` +
                          `**时间**: ${new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })}\n` +
                          `**状态**: 电压已低于安全阈值，请及时处理！`;
            } else if (data.alertType === 'high') {
              alertTitle = '⚠️ 电压过高告警';
              alertText = `### ⚠️ 电压过高告警\n\n` +
                          `**当前电压**: ${parseFloat(voltageValue).toFixed(3)}V\n` +
                          `**警告阈值**: ${parseFloat(data.thresholdValue).toFixed(3)}V\n` +
                          `**时间**: ${new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })}\n` +
                          `**状态**: 电压已超过安全阈值，请及时处理！`;
            }
            
            // 异步发送钉钉通知
            sendDingTalkMessage(alertTitle, alertText)
              .then(result => {
                logger.info('电压告警钉钉通知发送结果', { success: result !== null });
                
                // 返回alarm_received确认消息给小程序
                ws.send(JSON.stringify({
                  type: 'alarm_received',
                  success: true,
                  message: '电压告警已收到',
                  timestamp: Date.now(),
                  alertType: data.alertType,
                  voltage: voltageValue
                }));
              })
              .catch(error => {
                logger.error('发送电压告警钉钉通知时发生异常', {
                  error: error.message,
                  stack: error.stack
                });
                
                // 返回错误消息给小程序
                ws.send(JSON.stringify({
                  type: 'alarm_received',
                  success: false,
                  message: '发送钉钉通知失败',
                  timestamp: Date.now(),
                  error: error.message
                }));
              });
          } else {
            // 检查是否缺少电压参数或告警类型参数
            const missingParams = [];
            if (!voltageValue) {
              missingParams.push('vbat/bat');
            }
            if (!data.alertType) {
              missingParams.push('alertType');
            }
            
            const errorMsg = `缺少必要的voltage_alarm参数(${missingParams.join('或')})`;
            logger.warn(errorMsg, { clientId: clientId });
            ws.send(JSON.stringify({
              type: 'error',
              message: `Missing required parameters: ${missingParams.join(' and/or ')}`,
              timestamp: Date.now()
            }));
          }
          break;
          
        case 'get_history':
          logger.debug('处理get_history类型消息', { clientId: clientId });
          // 返回最近的电压历史数据给小程序
          ws.send(JSON.stringify({
            type: 'history_data',
            data: recentVoltageData,
            timestamp: Date.now(),
            totalRecords: recentVoltageData.length
          }));
          logger.info('返回历史电压数据给小程序', { clientId: clientId, recordCount: recentVoltageData.length });
          break;
          
        case 'clear_history':
          logger.debug('处理clear_history类型消息', { clientId: clientId });
          // 清空历史数据
          recentVoltageData.length = 0;
          
          try {
            // 尝试删除数据文件以确保完全清除
            if (fs.existsSync(DATA_FILE_PATH)) {
              fs.unlinkSync(DATA_FILE_PATH);
              logger.info('已删除历史数据文件', { filePath: DATA_FILE_PATH });
            }
            
            // 保险起见，再保存一次空数据
            saveVoltageDataToFile();
            logger.info('已确认清空历史数据', { clientId: clientId });
          } catch (error) {
            logger.error('清空历史数据文件时出错:', { error: error.message });
          }
          
          ws.send(JSON.stringify({
            type: 'history_cleared',
            success: true,
            message: '历史数据已清空',
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
          }));
          break;
          
        case 'get_device_logins':
          logger.debug('处理get_device_logins类型消息', { clientId: clientId });
          // 返回设备登录记录
          const loginRecords = Array.from(deviceLoginRecords.values()).sort((a, b) => b.loginTime - a.loginTime);
          ws.send(JSON.stringify({
            type: 'device_logins',
            data: loginRecords,
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
            totalRecords: loginRecords.length
          }));
          logger.info('返回设备登录记录给小程序', { clientId: clientId, recordCount: loginRecords.length });
          break;
          
        case 'get_important_messages':
          logger.debug('处理get_important_messages类型消息', { clientId: clientId });
          // 返回重要消息
          const messagesToReturn = [...importantMessages].sort((a, b) => b.timestamp - a.timestamp);
          ws.send(JSON.stringify({
            type: 'important_messages',
            data: messagesToReturn,
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
            totalRecords: messagesToReturn.length
          }));
          logger.info('返回重要消息给小程序', { clientId: clientId, recordCount: messagesToReturn.length });
          break;
          
        case 'save_important_message':
          logger.debug('处理save_important_message类型消息', { clientId: clientId });
          // 保存重要消息
          if (data.message && data.messageType) {
            const importantMsg = {
              type: data.messageType,
              timestamp: Date.now(),
              timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
              clientId: clientId,
              message: data.message,
              source: 'websocket',
              tags: data.tags || []
            };
            importantMessages.push(importantMsg);
            
            // 限制重要消息数量
            if (importantMessages.length > maxImportantMessages) {
              importantMessages.shift();
            }
            
            ws.send(JSON.stringify({
              type: 'message_saved',
              success: true,
              message: '重要消息已保存',
              timestamp: Date.now(),
              timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
            }));
            logger.info('重要消息已保存', { clientId: clientId, messageType: data.messageType });
          } else {
            ws.send(JSON.stringify({
              type: 'error',
              message: 'Missing required parameters: message and/or messageType',
              timestamp: Date.now(),
              timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
            }));
            logger.warn('缺少必要的save_important_message参数', { clientId: clientId });
          }
          break;
          
        case 'get_thresholds':
          logger.debug('处理get_thresholds类型消息', { clientId: clientId });
          // 返回当前电压阈值
          ws.send(JSON.stringify({
            type: 'thresholds_updated',
            data: CONFIG.VOLTAGE_THRESHOLDS,
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
          }));
          logger.info('返回当前电压阈值给小程序', { clientId: clientId });
          break;
          
        case 'get_system_config':
          logger.debug('处理get_system_config类型消息', { clientId: clientId });
          // 返回当前系统配置
          ws.send(JSON.stringify({
            type: 'system_config_updated',
            data: {
              alertCooldownTime: CONFIG.SYSTEM_CONFIG.ALERT_COOLDOWN_TIME,
              maxHistorySize: CONFIG.SYSTEM_CONFIG.maxHistorySize
            },
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
          }));
          logger.info('返回当前系统配置给小程序', { clientId: clientId });
          break;
          
        case 'get_all_configs':
          logger.debug('处理get_all_configs类型消息', { clientId: clientId });
          // 返回所有配置数据，包括阈值、webhook和关键字、警告冷静时间、数据条数等
          ws.send(JSON.stringify({
            type: 'all_configs_updated',
            data: {
              voltageThresholds: CONFIG.VOLTAGE_THRESHOLDS,
              dingtalkRobot: {
                webhook: CONFIG.DINGTALK_ROBOT.webhook,
                keywords: CONFIG.DINGTALK_ROBOT.keywords
              },
              systemConfig: {
                ALERT_COOLDOWN_TIME: CONFIG.SYSTEM_CONFIG.ALERT_COOLDOWN_TIME,
                maxHistorySize: CONFIG.SYSTEM_CONFIG.maxHistorySize
              },
              dataCounts: {
                recentVoltageDataCount: recentVoltageData.length, // 最近电压数据条数
                importantMessagesCount: importantMessages.length, // 重要消息条数
                deviceLoginRecordsCount: deviceLoginRecords.size, // 设备登录记录数
                maxHistorySize: CONFIG.SYSTEM_CONFIG.maxHistorySize // 最大历史记录数
              }
            },
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
          }));
          logger.info('返回所有配置数据给小程序', { clientId: clientId });
          break;
        
        case 'set_thresholds':
          logger.debug('处理set_thresholds类型消息', { clientId: clientId, receivedData: data });
          // 验证并更新电压阈值
          if (data.lower !== undefined && typeof data.lower === 'number' && data.lower > 0) {
            logger.info('准备更新低压阈值', { oldThreshold: CONFIG.VOLTAGE_THRESHOLDS.lower, newThreshold: data.lower, clientId: clientId });
            CONFIG.VOLTAGE_THRESHOLDS.lower = data.lower;
            logger.info('低压阈值已更新', { newThreshold: data.lower, clientId: clientId });
          } else {
            logger.warn('低压阈值参数无效', { lower: data.lower, clientId: clientId });
          }
          if (data.upper !== undefined && typeof data.upper === 'number' && data.upper > 0) {
            logger.info('准备更新高压阈值', { oldThreshold: CONFIG.VOLTAGE_THRESHOLDS.upper, newThreshold: data.upper, clientId: clientId });
            CONFIG.VOLTAGE_THRESHOLDS.upper = data.upper;
            logger.info('高压阈值已更新', { newThreshold: data.upper, clientId: clientId });
          } else {
            logger.warn('高压阈值参数无效', { upper: data.upper, clientId: clientId });
          }
          
          // 保存更新后的配置到文件
          try {
            const configPath = path.join(__dirname, 'config.json');
            // 更新configData对象
            configData.VOLTAGE_THRESHOLDS = CONFIG.VOLTAGE_THRESHOLDS;
            logger.debug('准备写入配置文件', { configData: configData });
            // 写入配置文件
            fs.writeFileSync(configPath, JSON.stringify(configData, null, 2), 'utf8');
            logger.info('配置已成功保存到文件', { configPath: configPath, savedThresholds: configData.VOLTAGE_THRESHOLDS });
          } catch (error) {
            logger.error('保存配置文件失败:', { error: error.message, stack: error.stack });
          }
          
          // 返回更新后的阈值给小程序
          ws.send(JSON.stringify({
            type: 'thresholds_updated',
            data: CONFIG.VOLTAGE_THRESHOLDS,
            success: true,
            message: '电压阈值已更新并保存',
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
          }));
          logger.info('电压阈值已更新并发送回小程序', { clientId: clientId, sentThresholds: CONFIG.VOLTAGE_THRESHOLDS });
          break;
          
        default:
          logger.warn(`未知消息类型 from client ${clientId}:`, { messageType: data.type });
          ws.send(JSON.stringify({
            type: 'error',
            message: `Unknown message type: ${data.type}`,
            timestamp: Date.now(),
            timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
          }));
      }
    } catch (error) {
      logger.error(`消息处理错误 for client ${clientId}:`, {
        error: error.message,
        stack: error.stack,
        messageSample: messageStr.substring(0, 100)
      });
      ws.send(JSON.stringify({
        type: 'error',
        message: 'Invalid message format',
        timestamp: Date.now(),
        errorType: error.name
      }));
    }
  });

  // 当收到MQTT服务器消息时，转发到小程序并检查电压
  mqttClient.on('message', (topic, message) => {
    logger.debug('收到MQTT消息', { topic: topic, messageLength: message.length });
    try {
      const messageStr = message.toString();
      logger.info(`接收到MQTT消息 on topic ${topic}`, {
        messageLength: messageStr.length,
        messageSample: messageStr.length > 100 ? messageStr.substring(0, 100) + '...' : messageStr,
        clientId: clientId
      });
      
      // 转发消息到小程序，处理可能的JSON字符串
      try {
        // 尝试解析messageStr，检查它是否已经是JSON字符串
        const parsedPayload = JSON.parse(messageStr);
        // 如果是JSON对象，直接转发解析后的数据，避免双重序列化
        ws.send(JSON.stringify({
          type: 'message',
          topic: topic,
          payload: parsedPayload
        }));
      } catch (parseError) {
        // 如果不是有效的JSON字符串，作为普通字符串处理
        ws.send(JSON.stringify({
          type: 'message',
          topic: topic,
          payload: messageStr
        }));
      }
      
      logger.info(`已将消息转发给WebSocket客户端`, { clientId: clientId, topic: topic });
      
      // 检查是否为电压主题并进行告警检查
      if (topic === '/robot/zack/vbat') {
        try {
          // 尝试解析消息中的电压数据
          const data = JSON.parse(messageStr);
          
          // 检查是否为测试消息
          if (data.voltage === 'test' || data.vbat === 'test' || data.type === 'link_test' || data.testMessageId) {
            logger.info('接收到测试消息，准备回复', { testMessageId: data.testMessageId || data.id });
            
            // 添加到重要消息列表
            const importantMsg = {
              type: 'test_message',
              timestamp: Date.now(),
              timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
              clientId: clientId,
              message: data,
              source: 'mqtt'
            };
            importantMessages.push(importantMsg);
            
            // 限制重要消息数量
            if (importantMessages.length > maxImportantMessages) {
              importantMessages.shift();
            }
            
            // 发送测试响应
            const testResponse = JSON.stringify({
              type: 'test_response',
              success: true,
              message: '链路测试成功',
              timestamp: Date.now(),
              timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
              vbat: 0, // 添加vbat字段以便小程序能正确处理
              testMessageId: data.testMessageId || data.id || 'default_test_id', // 包含原始测试消息ID
              debugInfo: {
                clientId: clientId,
                activeConnections: activeConnections.size,
                broker: 'mqtt://broker.emqx.io',
                timestamp: Date.now()
              }
            });
              // 直接通过WebSocket将测试响应发送回小程序
              ws.send(testResponse);
              logger.info('收到测试消息并直接通过WebSocket发送响应', { clientId: clientId });
              
              // 同时也发布到MQTT主题，保持兼容性
              mqttClient.publish('/robot/zack/vbat', testResponse, (err) => {
                  if (err) {
                    logger.error('发布测试响应到MQTT主题失败:', { error: err.message });
                  } else {
                    logger.info('成功发布测试响应到MQTT主题', { responseSize: testResponse.length });
                  }
                });
              }
           
          // 检查是否为阈值设置消息
          if (data.type === 'set_thresholds' || data.setThresholds === true || data.VOLTAGE_THRESHOLDS) {
            try {
              // 保存原始阈值用于记录变更
              const oldThresholds = { ...CONFIG.VOLTAGE_THRESHOLDS };
              
              // 处理不同格式的阈值数据
              let lowerValue = null;
              let upperValue = null;
              
              // 格式1: 直接在顶层data对象中包含lower和upper字段
              if (typeof data.lower === 'number' && data.lower > 0) {
                lowerValue = data.lower;
              }
              if (typeof data.upper === 'number' && data.upper > 0) {
                upperValue = data.upper;
              }
              
              // 格式2: 在lowerThreshold和upperThreshold字段中
              if (typeof data.lowerThreshold === 'number' && data.lowerThreshold > 0) {
                lowerValue = data.lowerThreshold;
              }
              if (typeof data.upperThreshold === 'number' && data.upperThreshold > 0) {
                upperValue = data.upperThreshold;
              }
              
              // 格式3: 在VOLTAGE_THRESHOLDS对象中
              if (data.VOLTAGE_THRESHOLDS && typeof data.VOLTAGE_THRESHOLDS === 'object') {
                if (typeof data.VOLTAGE_THRESHOLDS.lower === 'number' && data.VOLTAGE_THRESHOLDS.lower > 0) {
                  lowerValue = data.VOLTAGE_THRESHOLDS.lower;
                }
                if (typeof data.VOLTAGE_THRESHOLDS.upper === 'number' && data.VOLTAGE_THRESHOLDS.upper > 0) {
                  upperValue = data.VOLTAGE_THRESHOLDS.upper;
                }
              }
              
              // 验证并更新电压阈值
              if (lowerValue !== null) {
                CONFIG.VOLTAGE_THRESHOLDS.lower = lowerValue;
                logger.info('低压阈值已更新', { oldThreshold: oldThresholds.lower, newThreshold: lowerValue });
              }
              if (upperValue !== null) {
                CONFIG.VOLTAGE_THRESHOLDS.upper = upperValue;
                logger.info('高压阈值已更新', { oldThreshold: oldThresholds.upper, newThreshold: upperValue });
              }
              
              // 保存更新后的配置到文件
              try {
                const configPath = path.join(__dirname, 'config.json');
                // 更新configData对象
                configData.VOLTAGE_THRESHOLDS = CONFIG.VOLTAGE_THRESHOLDS;
                // 写入配置文件
                fs.writeFileSync(configPath, JSON.stringify(configData, null, 2), 'utf8');
                logger.info('配置已成功保存到文件', { configPath: configPath, savedThresholds: configData.VOLTAGE_THRESHOLDS });
              } catch (error) {
                logger.error('保存配置文件失败:', { error: error.message });
              }
              
              // 发送确认消息到MQTT主题
              const response = JSON.stringify({
                type: 'thresholds_updated',
                success: true,
                message: '电压阈值已更新并保存',
                timestamp: Date.now(),
                timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
                data: CONFIG.VOLTAGE_THRESHOLDS,
                oldThresholds: oldThresholds,
                clientId: clientId || 'mqtt_client'
              });
              
              // 发布确认消息到MQTT主题
              mqttClient.publish('/robot/zack/vbat', response, (err) => {
                if (err) {
                  logger.error('发布阈值更新确认消息到MQTT主题失败:', { error: err.message });
                } else {
                  logger.info('成功发布阈值更新确认消息到MQTT主题', { responseSize: response.length });
                }
              });
              
              // 将确认消息发送给所有连接的WebSocket客户端
              activeConnections.forEach((connection, id) => {
                if (connection.ws && connection.ws.readyState === WebSocket.OPEN) {
                  connection.ws.send(response);
                  logger.info('已通过WebSocket发送阈值更新确认消息', { clientId: id });
                }
              });
            } catch (error) {
              logger.error('更新电压阈值时发生错误:', { error: error.message, stack: error.stack, data: data });
              
              // 发送错误消息
              const errorResponse = JSON.stringify({
                type: 'error',
                message: '更新阈值时发生错误: ' + error.message,
                timestamp: Date.now(),
                timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' })
              });
              
              // 发布错误消息到MQTT主题
              mqttClient.publish('/robot/zack/vbat', errorResponse, (err) => {
                if (err) {
                  logger.error('发布阈值更新错误消息到MQTT主题失败:', { error: err.message });
                }
              });
            }
          }
          
          // 检查是否为钉钉机器人配置更新消息
          if (data.type === 'set_dingtalk_config' || data.setDingtalkConfig === true) {
            try {
              // 验证并更新钉钉机器人配置
              if (data.webhook && typeof data.webhook === 'string' && data.webhook.startsWith('https://oapi.dingtalk.com/robot/send')) {
                // 去除webhook字符串前后可能存在的空格或其他字符
                const cleanWebhook = data.webhook.trim();
                CONFIG.DINGTALK_ROBOT.webhook = cleanWebhook;
                // 同步更新DINGTALK_ROBOT_CONFIG引用
                DINGTALK_ROBOT_CONFIG.webhook = cleanWebhook;
                logger.info('钉钉机器人webhook已更新', { webhookSample: cleanWebhook.substring(0, 50) + '...' });
              }
              if (Array.isArray(data.keywords)) {
                CONFIG.DINGTALK_ROBOT.keywords = data.keywords;
                // 同步更新DINGTALK_ROBOT_CONFIG引用
                DINGTALK_ROBOT_CONFIG.keywords = data.keywords;
                logger.info('钉钉机器人keywords已更新', { keywords: data.keywords });
              }
              
              // 保存配置变更记录到重要消息
            const configChangeMsg = {
              type: 'dingtalk_config_change',
              timestamp: Date.now(),
              timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
              clientId: clientId,
              message: '钉钉机器人配置已更新',
              source: 'mqtt',
              oldConfig: { ...oldConfig },
              newConfig: {
                webhook: CONFIG.DINGTALK_ROBOT.webhook,
                keywords: CONFIG.DINGTALK_ROBOT.keywords
              }
            };
              importantMessages.push(configChangeMsg);
              
              // 限制重要消息数量
              if (importantMessages.length > maxImportantMessages) {
                importantMessages.shift();
              }
              
              // 发送确认消息到专门的MQTT主题
              const response = JSON.stringify({
                type: 'dingtalk_config_updated',
                success: true,
                message: '钉钉机器人配置已更新',
                timestamp: Date.now(),
                timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
                newConfig: {
                  webhook: CONFIG.DINGTALK_ROBOT.webhook,
                  keywords: CONFIG.DINGTALK_ROBOT.keywords
                },
                oldConfig: oldConfig || {},
                clientId: clientId,
                configType: 'dingtalk_robot'
              });
              
              // 发布到专门的响应主题
              const responseTopic = '/robot/zack/vbat/config_response';
              mqttClient.publish(responseTopic, response, (err) => {
                if (err) {
                  logger.error('发布钉钉机器人配置更新确认消息失败:', { error: err.message, topic: responseTopic });
                } else {
                  logger.info('成功发布钉钉机器人配置更新确认消息到专门主题', { topic: responseTopic });
                }
              });
              
              // 为保持兼容性，同时发布到原主题
              mqttClient.publish('/robot/zack/vbat', response, (err) => {
                if (err) {
                  logger.error('发布钉钉机器人配置更新确认消息到兼容性主题失败:', { error: err.message });
                }
              });
              
              // 通过WebSocket通知客户端
              ws.send(JSON.stringify({
                type: 'dingtalk_config_updated',
                success: true,
                message: '钉钉机器人配置已通过MQTT更新',
                timestamp: Date.now(),
                newConfig: {
                  webhook: CONFIG.DINGTALK_ROBOT.webhook,
                  keywords: CONFIG.DINGTALK_ROBOT.keywords
                }
              }));
              
            } catch (error) {
              logger.error('更新钉钉机器人配置时发生错误:', { error: error.message, stack: error.stack });
            }
          }
          
          // 检查是否为系统配置更新消息
          if (data.type === 'set_system_config' || data.setSystemConfig === true) {
            try {
              // 首先通过WebSocket返回设置成功的响应（如果是从WebSocket连接发起的请求）
              ws.send(JSON.stringify({
                type: 'system_config_update_response',
                success: true,
                message: '系统配置更新请求已收到，正在处理',
                timestamp: Date.now()
              }));
              
              // 为从MQTTx等外部MQTT客户端发送的配置命令添加专门的回复
              // 生成一个唯一的请求ID，用于匹配请求和响应
              const requestId = data.requestId || `req_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
              // 保存原始配置用于记录变更
              const oldSystemConfig = { ...CONFIG.SYSTEM_CONFIG };
              
              // 验证并更新系统配置
              if (typeof data.alertCooldownTime === 'number' && data.alertCooldownTime >= 0) {
                CONFIG.SYSTEM_CONFIG.ALERT_COOLDOWN_TIME = data.alertCooldownTime;
                // 同步更新全局变量
                ALERT_COOLDOWN_TIME = data.alertCooldownTime;
                logger.info('告警冷却时间已更新', { newTime: data.alertCooldownTime });
              }
              if (typeof data.maxHistorySize === 'number' && data.maxHistorySize > 0 && data.maxHistorySize <= 1000) {
                CONFIG.SYSTEM_CONFIG.maxHistorySize = data.maxHistorySize;
                // 同步更新全局变量
                maxHistorySize = data.maxHistorySize;
                // 立即应用新的历史记录大小限制
                while (recentVoltageData.length > maxHistorySize) {
                  recentVoltageData.shift();
                }
                logger.info('最大历史记录数已更新', { newSize: data.maxHistorySize });
              }
              
              // 保存配置变更记录到重要消息
              const systemConfigChangeMsg = {
                type: 'system_config_change',
                timestamp: Date.now(),
                timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
                clientId: clientId,
                message: `系统配置已更新`,
                source: 'mqtt',
                oldConfig: { ...oldSystemConfig },
                newConfig: { ...CONFIG.SYSTEM_CONFIG }
              };
              
              // 持久化系统配置到配置文件
              try {
                const configPath = path.join(__dirname, 'config.json');
                const updatedConfig = {
                  ...configData,
                  SYSTEM_CONFIG: CONFIG.SYSTEM_CONFIG
                };
                fs.writeFileSync(configPath, JSON.stringify(updatedConfig, null, 2));
                logger.info('系统配置已持久化到配置文件', { configPath: configPath });
              } catch (error) {
                logger.error('持久化系统配置失败:', { error: error.message });
              }
              importantMessages.push(systemConfigChangeMsg);
              
              // 限制重要消息数量
              if (importantMessages.length > maxImportantMessages) {
                importantMessages.shift();
              }
              
              // 发送确认消息到专门的MQTT主题
              const systemResponse = JSON.stringify({
                type: 'system_config_updated',
                success: true,
                message: '系统配置已更新',
                timestamp: Date.now(),
                timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
                newConfig: { ...CONFIG.SYSTEM_CONFIG },
                oldConfig: oldSystemConfig,
                clientId: clientId,
                configType: 'system_config'
              });
              
              // 发布到专门的响应主题
              const responseTopic = '/robot/zack/vbat/config_response';
              mqttClient.publish(responseTopic, systemResponse, (err) => {
                if (err) {
                  logger.error('发布系统配置更新确认消息失败:', { error: err.message, topic: responseTopic });
                } else {
                  logger.info('成功发布系统配置更新确认消息到专门主题', { topic: responseTopic });
                }
              });
              
              // 为保持兼容性，同时发布到原主题
              mqttClient.publish('/robot/zack/vbat', systemResponse, (err) => {
                if (err) {
                  logger.error('发布系统配置更新确认消息到兼容性主题失败:', { error: err.message });
                }
              });
              
              // 通过WebSocket通知客户端，包含详细的更新信息和配置文件保存状态
              ws.send(JSON.stringify({
                type: 'system_config_updated',
                success: true,
                message: '系统配置已成功更新并保存',
                timestamp: Date.now(),
                timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
                newConfig: { ...CONFIG.SYSTEM_CONFIG },
                oldConfig: oldSystemConfig,
                savedToFile: true,
                updatedFields: Object.keys(data).filter(key => 
                  ['alertCooldownTime', 'maxHistorySize'].includes(key) && 
                  data[key] !== oldSystemConfig[key]
                )
              }));
              
              // 发送专门的回复消息给MQTT客户端（如MQTTx），明确表示配置已成功保存
              // 这个回复消息将包含原始请求的ID，便于客户端匹配请求和响应
              const mqttClientResponse = JSON.stringify({
                type: 'system_config_save_response',
                success: true,
                message: '系统配置已成功更新并保存到文件',
                requestId: requestId,
                timestamp: Date.now(),
                timestampBeijing: new Date().toLocaleString('zh-CN', { timeZone: 'Asia/Shanghai' }),
                configType: 'system',
                updatedConfig: {
                  alertCooldownTime: CONFIG.SYSTEM_CONFIG.ALERT_COOLDOWN_TIME,
                  maxHistorySize: CONFIG.SYSTEM_CONFIG.maxHistorySize
                },
                updatedFields: Object.keys(data).filter(key => 
                  ['alertCooldownTime', 'maxHistorySize'].includes(key) && 
                  data[key] !== oldSystemConfig[key]
                )
              });
              
              // 发布回复消息到响应主题
              mqttClient.publish('/robot/zack/vbat/config_response', mqttClientResponse, (err) => {
                if (err) {
                  logger.error('发布系统配置保存确认消息失败:', { error: err.message });
                } else {
                  logger.info('成功发布系统配置保存确认消息到MQTT客户端', { requestId: requestId });
                }
              });
              
              // 为保持兼容性，同时发布到原主题
              mqttClient.publish('/robot/zack/vbat', mqttClientResponse);
              
            } catch (error) {
              logger.error('更新系统配置时发生错误:', { error: error.message, stack: error.stack });
              
              // 发送错误回复给MQTT客户端
              const errorResponse = JSON.stringify({
                type: 'system_config_save_response',
                success: false,
                message: '更新系统配置失败: ' + error.message,
                requestId: requestId,
                timestamp: Date.now(),
                errorDetails: error.message,
                configType: 'system'
              });
              
              // 发布错误消息到响应主题
              mqttClient.publish('/robot/zack/vbat/config_response', errorResponse);
              // 同时发布到原主题保持兼容性
              mqttClient.publish('/robot/zack/vbat', errorResponse);
            }
          }
              

          
          // 增强的数据处理：尝试从各种可能的字段中提取电压数据
          const voltageData = extractVoltageData(data);
          if (voltageData.vbat !== null && !isNaN(voltageData.vbat)) {
            logger.debug('成功提取电压数据', {
              voltage: voltageData.vbat,
              sourceField: voltageData.sourceField,
              clientId: clientId
            });
            
            // 保存电压数据到历史记录
            const voltageRecord = {
              vbat: voltageData.vbat,
              timestamp: Date.now(),
              from: 'mqtt',
              sourceField: voltageData.sourceField,
              // 保存原始数据的所有字段，便于调试和分析
              originalData: {...data},
              formattedTime: new Date().toLocaleString('zh-CN')
            };
            recentVoltageData.push(voltageRecord);
            
            // 保持历史数据量不超过最大限制
            if (recentVoltageData.length > maxHistorySize) {
              recentVoltageData.shift();
            }
            
            // 按需保存数据 - 在添加新数据后立即保存
            saveVoltageDataToFile();
            
            // 记录电压日志
            logger.info(`记录电压数据: ${voltageData.vbat.toFixed(3)}V (来自${voltageData.sourceField}字段)`, {
              clientId: clientId,
              voltage: voltageData.vbat,
              historyCount: recentVoltageData.length
            });
            
            // 异步检查电压告警
            checkVoltageAlert(voltageData.vbat);
          } else if (!(data.type === 'set_dingtalk_config' || data.setDingtalkConfig === true)) {
            // 即使没有找到电压数据，也记录接收到的消息，便于调试
            logger.warn(`收到/robot/zack/vbat主题消息，但未找到有效电压数据`, {
              dataSample: JSON.stringify(data).substring(0, 100),
              clientId: clientId
            });
          }
        } catch (parseError) {
          logger.error(`解析电压消息失败:`, {
            error: parseError.message,
            stack: parseError.stack,
            messageSample: messageStr.substring(0, 50),
            clientId: clientId
          });
          // 即使解析失败，也继续处理，因为消息已经成功转发到小程序
        }
        
      }
    } catch (error) {
      logger.error(`转发MQTT消息到客户端失败`, {
        clientId: clientId,
        error: error.message,
        stack: error.stack,
        topic: topic
      });
    }
  });

  // 处理MQTT错误
  mqttClient.on('error', (error) => {
    logger.error(`MQTT error for client ${clientId}:`, {
      error: error.message,
      stack: error.stack,
      code: error.code,
      address: error.address,
      port: error.port
    });
    ws.send(JSON.stringify({
      type: 'mqtt_error',
      error: error.message || 'Unknown MQTT error',
      clientId: clientId,
      broker: mqttBroker,
      timestamp: Date.now(),
      errorType: error.name || error.code || 'unknown'
    }));
  });
  
  // 处理MQTT连接超时
  mqttClient.on('timeout', () => {
    logger.error(`MQTT connection timeout for client ${clientId}:`, { broker: mqttBroker });
    ws.send(JSON.stringify({
      type: 'mqtt_timeout',
      error: 'Connection to MQTT server timed out',
      clientId: clientId,
      broker: mqttBroker
    }));
  });

  // 处理MQTT断开连接
  mqttClient.on('close', () => {
    logger.info('MQTT connection closed', {
      clientId: clientId,
      activeConnections: activeConnections.size,
      broker: mqttBroker
    });
    ws.send(JSON.stringify({
      type: 'mqtt_disconnected',
      message: 'MQTT server connection closed'
    }));
    
    // 启动重连逻辑
    if (!reconnecting) {
      logger.info('尝试重新连接MQTT服务器...', {
        clientId: clientId,
        broker: mqttBroker
      });
      reconnectMQTT();
    }
  });
  
  // MQTT重连函数
  function reconnectMQTT() {
    if (!activeConnections.has(clientId)) {
      logger.info(`Client ${clientId} no longer active, stopping reconnection attempts`);
      reconnecting = false;
      return;
    }
    
    // 标记为正在重连
    reconnecting = true;
    logger.info(`尝试重新连接MQTT服务器... 客户端: ${clientId}`);
    
    const newMqttClient = mqtt.connect(mqttBroker, {
      clientId: `cloud-client-${clientId}`,
      clean: true,
      connectTimeout: 5000,
      keepalive: 60
    });
    
    // 添加事件处理程序
    newMqttClient.on('connect', () => {
      console.log(`MQTT重连成功 for client: ${clientId}`);
      // 更新mqttClient引用
      mqttClient = newMqttClient;
      
      // 恢复订阅
      mqttClient.subscribe('/robot/zack/vbat');
      
      // 通知客户端重连成功
      ws.send(JSON.stringify({
        type: 'mqtt_reconnected',
        message: 'MQTT server reconnected',
        clientId: clientId,
        reconnectTime: new Date().toISOString()
      }));
      
      // 重置重连状态
      reconnecting = false;
    });
    
    // 消息处理逻辑
    newMqttClient.on('message', (topic, message) => {
      try {
        const messageStr = message.toString();
        // 处理可能的JSON字符串，避免双重序列化
        try {
          // 尝试解析messageStr，检查它是否已经是JSON字符串
          const parsedPayload = JSON.parse(messageStr);
          // 如果是JSON对象，直接转发解析后的数据
          const messageData = {
            type: 'message',
            topic: topic,
            payload: parsedPayload
          };
          ws.send(JSON.stringify(messageData));
        } catch (parseError) {
          // 如果不是有效的JSON字符串，作为普通字符串处理
          const messageData = {
            type: 'message',
            topic: topic,
            payload: messageStr
          };
          ws.send(JSON.stringify(messageData));
        }
        logger.info(`已将消息转发给WebSocket客户端: ${clientId}`, { topic: topic });
        
        // 检查是否为电压主题并进行告警检查
        if (topic === '/robot/zack/vbat') {
          try {
            // 尝试解析消息中的电压数据
            const data = JSON.parse(messageStr);
            
            // 检查是否为测试消息
            if (data.voltage === 'test' || data.vbat === 'test' || data.type === 'link_test' || data.testMessageId) {
              // 发送测试响应
              const testResponse = JSON.stringify({
                type: 'test_response',
                success: true,
                message: '链路测试成功',
                timestamp: Date.now(),
                vbat: 0, // 添加vbat字段以便小程序能正确处理
                testMessageId: data.testMessageId || data.id || 'default_test_id', // 包含原始测试消息ID
                debugInfo: {
                  clientId: clientId,
                  activeConnections: activeConnections.size,
                  broker: 'mqtt://broker.emqx.io',
                  timestamp: Date.now()
                }
              });
              newMqttClient.publish('/robot/zack/vbat', testResponse);
              logger.info('收到测试消息并发送响应', { clientId: clientId, testMessageId: data.testMessageId || data.id });
            }
            
            // 增强的数据处理：尝试从各种可能的字段中提取电压数据
            const voltageData = extractVoltageData(data);
            if (voltageData.vbat !== null && !isNaN(voltageData.vbat)) {
              // 保存电压数据到历史记录
              const voltageRecord = {
                vbat: voltageData.vbat,
                timestamp: Date.now(),
                from: 'mqtt',
                sourceField: voltageData.sourceField,
                // 保存原始数据的所有字段，便于调试和分析
                originalData: {...data},
                formattedTime: new Date().toLocaleString('zh-CN')
              };
              recentVoltageData.push(voltageRecord);
              
              // 保持历史数据量不超过最大限制
              if (recentVoltageData.length > maxHistorySize) {
                recentVoltageData.shift();
              }
              
              // 按需保存数据 - 在添加新数据后立即保存
              saveVoltageDataToFile();
              
              // 记录电压日志
              logger.info(`记录电压数据: ${voltageData.vbat.toFixed(3)}V (来自${voltageData.sourceField}字段), 历史记录数: ${recentVoltageData.length}`);
              
              // 异步检查电压告警
              checkVoltageAlert(voltageData.vbat);
            } else {
              // 即使没有找到电压数据，也记录接收到的消息，便于调试
              logger.debug(`收到/robot/zack/vbat主题消息，但未找到有效电压数据:`, data);
            }
          } catch (parseError) {
            logger.error(`解析电压消息失败:`, { error: parseError.message, data: data });
            // 即使解析失败，也继续处理，因为消息已经成功转发到小程序
          }
        }
      } catch (error) {
        logger.error(`Failed to forward MQTT message to client ${clientId}:`, { error: error.message });
      }
    });
    
    // 错误处理逻辑
    newMqttClient.on('error', (error) => {
      logger.error(`MQTT error for client ${clientId}:`, { error: error.message });
      ws.send(JSON.stringify({
        type: 'mqtt_error',
        error: error.message || 'Unknown MQTT error'
      }));
    });
    
    // 关闭处理逻辑
    newMqttClient.on('close', () => {
      logger.info(`MQTT重连尝试失败，将在5秒后再次尝试`);
      // 重置重连状态
      reconnecting = false;
      // 继续重连逻辑
      setTimeout(reconnectMQTT, 5000);
    });
  }

  // 处理WebSocket关闭
  ws.on('close', (code, reason) => {
    logger.info(`Client disconnected: ${clientId}`, {
      previousConnections: activeConnections.size,
      newConnections: activeConnections.size - 1,
      code: code || 'unknown',
      reason: reason ? reason.toString() : 'unknown'
    });
    activeConnections.delete(clientId);
    
    // 清除连接超时定时器
    clearTimeout(connectionTimeout);
    
    // 清除心跳定时器
    if (heartbeatInterval) {
      clearInterval(heartbeatInterval);
      logger.debug('已清除客户端心跳定时器', { clientId: clientId });
    }
    
    // 构建下线消息
    const offlineMessage = {
      type: 'device_offline',
      device: 'mini_program',
      timestamp: new Date().toISOString(),
      clientId: clientId,
      disconnectCode: code || 'unknown',
      disconnectReason: reason ? reason.toString() : 'unknown'
    };
    
    // 发布下线消息到MQTT主题
    if (mqttClient.connected) {
      mqttClient.publish('/robot/zack/vbat', JSON.stringify(offlineMessage), (err) => {
        if (err) {
          logger.error('发布下线消息失败:', { error: err.message, clientId: clientId });
        } else {
          logger.info('成功发布下线消息到MQTT主题: /robot/zack/vbat', { clientId: clientId });
        }
      });
    }
    
    // 关闭对应的MQTT连接
    if (mqttClient.connected) {
      mqttClient.end();
    }
  });
  
  // 设置心跳机制
  const heartbeatInterval = setInterval(() => {
    try {
      // 检查WebSocket连接状态
      if (ws.readyState === WebSocket.OPEN) {
        // 发送心跳消息给客户端
        ws.send(JSON.stringify({
          type: 'heartbeat',
          timestamp: Date.now(),
          status: {
            connected: true,
            mqttConnected: mqttClient && mqttClient.connected,
            clientId: clientId,
            activeConnections: activeConnections.size
          }
        }));
        
        // 同时检查MQTT连接状态
        if (mqttClient && !mqttClient.connected && !reconnecting) {
          logger.warn(`MQTT connection lost, attempting to reconnect for client: ${clientId}`);
          reconnectMQTT();
        }
      }
    } catch (error) {
      logger.error(`Heartbeat error for client ${clientId}:`, { error: error.message });
    }
  }, 30000); // 每30秒发送一次心跳
  
  // 重连状态标记
  let reconnecting = false;

  // 处理WebSocket错误
  ws.on('error', (error) => {
    logger.error(`WebSocket错误 for client ${clientId}:`, {
      error: error.message,
      stack: error.stack,
      code: error.code,
      address: error.address,
      port: error.port
    });
  });
});

// 启动服务器
const PORT = process.env.PORT || 80;
server.listen(PORT, () => {
  logger.info(`MQTT proxy server is running on port ${PORT}`);
});

// 处理进程终止信号
process.on('SIGINT', () => {
  logger.info('Server is shutting down...');
  server.close(() => {
    logger.info('Server closed');
    process.exit(0);
  });
});

process.on('unhandledRejection', (reason, promise) => {
  logger.error('Unhandled Rejection at:', { promise: promise, reason: reason });
});